package com.sunzm.dynamic_rule.functions;

import com.alibaba.fastjson.JSON;
import com.sunzm.dynamic_rule.pojo.*;
import com.sunzm.dynamic_rule.service.QueryRouterV4;
import com.sunzm.dynamic_rule.utils.RuleOperationHandler2;
import com.sunzm.dynamic_rule.utils.StateDescUtil;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * 规则核心处理函数V5
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-07-07 1:05
 */
public class RuleProcessFunctionV5 extends KeyedBroadcastProcessFunction<String, LogBean, String, ResultBean> {
    private static final Logger logger = LoggerFactory.getLogger(RuleProcessFunctionV5.class);
    private QueryRouterV4 queryRouter;

    private ListState<LogBean> eventState;

    @Override
    public void open(Configuration parameters) throws Exception {

        /*
         * 准备一个存储明细事件的state
         * 控制state的ttl周期为最近2小时
         */

        ListStateDescriptor<LogBean> eventStateDesc = StateDescUtil.eventStateDesc;
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).updateTtlOnCreateAndWrite().build();
        eventStateDesc.enableTimeToLive(ttlConfig);
        eventState = getRuntimeContext().getListState(eventStateDesc);

        // 构造一个查询路由控制器
        queryRouter = new QueryRouterV4(eventState);
    }


    @Override
    public void processElement(LogBean logBean, ReadOnlyContext ctx, Collector<ResultBean> out) throws Exception {
        ReadOnlyBroadcastState<String, RuleStateBean2> ruleState = ctx.getBroadcastState(StateDescUtil.ruleKieStateDesc);

        // 将收到的事件放入历史明细state存储中
        // 超过2小时的logBean会被自动清除（前面设置了ttl存活时长）
        eventState.add(logBean);

        Iterable<Map.Entry<String, RuleStateBean2>> entries = ruleState.immutableEntries();

        for (Map.Entry<String, RuleStateBean2> entry : entries) {
            RuleStateBean2 stateBean = entry.getValue();

            // 获取ruleparam对象
            RuleParam ruleParam = stateBean.getRuleParam();

            /*
             * 主逻辑，进行规则触发和计算
             */
            if (ruleParam.getTriggerParam().getEventId().equals(logBean.getEventId())) {
                logger.debug("规则:{},用户:{},触发事件:{},触发时间:{}", ruleParam.getRuleName(),logBean.getDeviceId(),logBean.getEventId(),logBean.getTimeStamp());

                boolean b1 = queryRouter.profileQuery(logBean, ruleParam);
                if(!b1){
                    return;
                }

                boolean b2 = queryRouter.sequenceConditionQuery(logBean, ruleParam);
                if(!b2) {
                    return;
                }

                boolean b3 = queryRouter.countConditionQuery(logBean, ruleParam);
                if(!b3) {
                    return;
                }


                // 输出一个规则匹配成功的结果
                ResultBean resultBean = new ResultBean();
                resultBean.setTimeStamp(logBean.getTimeStamp());
                resultBean.setRuleId(ruleParam.getRuleName());
                resultBean.setDeviceId(logBean.getDeviceId());
                logger.info("{}规则,触发人:{},计算匹配成功", ruleParam.getRuleName(),logBean.getDeviceId());

                out.collect(resultBean);
            }
        }
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<ResultBean> out) {
        BroadcastState<String, RuleParam> mapState = ctx.getBroadcastState(StateDescUtil.ruleParaStateDesc);

        // 解析json成对象
        RuleCanalBean ruleCanalBean = JSON.parseObject(value, RuleCanalBean.class);
        logger.info("收到一个规则库的操作,信息为: {}", ruleCanalBean);

        // 分情况处理获取的规则操作信息（新增，更新，删除，停用，启用）
        RuleOperationHandler2.handleRuleOper(ruleCanalBean, mapState);
    }
}
